package com.yanfan.mqtt.handler;

import com.alibaba.fastjson2.JSONObject;
import com.yanfan.base.service.ISessionStore;
import com.yanfan.base.session.Session;
import com.yanfan.base.util.AttributeUtils;
import com.yanfan.common.constant.YanfanConstant;
import com.yanfan.common.core.mq.DeviceReportBo;
import com.yanfan.common.core.mq.DeviceStatusBo;
import com.yanfan.common.core.redis.RedisCache;
import com.yanfan.common.enums.DeviceStatus;
import com.yanfan.common.enums.ServerType;
import com.yanfan.common.enums.TopicType;
import com.yanfan.common.utils.DateUtils;
import com.yanfan.common.utils.StringUtils;
import com.yanfan.common.utils.gateway.mq.TopicsUtils;
import com.yanfan.iot.domain.Device;
import com.yanfan.iot.model.DeviceMqttConnectVO;
import com.yanfan.iot.ruleEngine.MsgContext;
import com.yanfan.iot.ruleEngine.RuleProcess;
import com.yanfan.iot.service.IDeviceService;
import com.yanfan.mq.redischannel.producer.MessageProducer;
import com.yanfan.mqtt.annotation.Process;
import com.yanfan.mqtt.handler.adapter.MqttHandler;
import com.yanfan.mqtt.manager.ClientManager;
import com.yanfan.mqtt.manager.ResponseManager;
import com.yanfan.mqtt.manager.RetainMsgManager;
import com.yanfan.mqtt.manager.SessionManger;
import com.yanfan.mqtt.model.ClientMessage;
import com.yanfan.mqtt.service.IMessageStore;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.*;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

/**
 * 客户端消息推送处理类
 *
 * @author bill
 */
@Slf4j
@Process(type = MqttMessageType.PUBLISH)
public class MqttPublish implements MqttHandler {

    @Autowired
    private IMessageStore messageStore;
    @Resource
    private TopicsUtils topicsUtils;
    @Resource
    private RedisCache redisCache;
    @Resource
    private RuleProcess ruleProcess;
    @Resource
    private ISessionStore sessionStore;
    @Resource
    private IDeviceService deviceService;

    @Override
    public void handler(ChannelHandlerContext ctx, MqttMessage message) {
        MqttPublishMessage publishMessage = (MqttPublishMessage) message;
        /*获取客户端id*/
        String clientId = AttributeUtils.getClientId(ctx.channel());
        String topicName = publishMessage.variableHeader().topicName();
        log.debug("=>***客户端【{}】,主题【{}】,推送消息【{}】", clientId, topicName, ByteBufUtil.hexDump(publishMessage.content()));
        MqttYllj(ctx, clientId, publishMessage);
        DeviceMqttConnectVO deviceMqttConnect = deviceService.getDeviceMqttConnect(clientId, null, null);
        String deviceNumber;
        if (deviceMqttConnect == null) {
            deviceNumber = topicsUtils.parseSerialNumber(topicName);
            if (Objects.nonNull(deviceNumber)) {
                deviceMqttConnect = deviceService.getDeviceMqttConnect(null, deviceNumber, null);
            }
        }
        // 以get结尾是模拟客户端数据,只转发消息
        if (topicName.endsWith(YanfanConstant.MQTT.PROPERTY_GET_SIMULATE)) {
            log.debug("消息处理模式：模拟主题");
            sendTestToMQ(publishMessage);
        }
        //设备mqtt连接配置不为空，mqttstats=1代表是自定义的主题，走自定义主题逻辑
        else if (Objects.nonNull(deviceMqttConnect) && deviceMqttConnect.getMqttstats() == 1) {
            log.debug("消息处理模式：自定义主题");
            //自定义发布主题
            if (topicName.equals(deviceMqttConnect.getFpname())) {
                log.debug("开始处理自定义发布主题");
                String snumber = deviceMqttConnect.getSnumber();
                // 将有人上报的消息转发到我们的平台
                Device device = deviceService.selectDeviceBySerialNumber(snumber);
                deviceNumber = device.getSerialNumber();
                Long productId = device.getProductId();
                String topic = "/" + productId + "/" + deviceNumber + "/property/post";

                MqttQoS qos = publishMessage.fixedHeader().qosLevel();
                ByteBuf sendBuf = publishMessage.content().retainedDuplicate();
                sendBuf.resetReaderIndex();
                /*配置推送消息类型*/
                MqttFixedHeader Header = new MqttFixedHeader(MqttMessageType.PUBLISH, false, qos, publishMessage
                        .fixedHeader()
                        .isRetain(), 0);
                /*设置topic packetId*/
                MqttPublishVariableHeader publishVariableHeader = new MqttPublishVariableHeader(topic, publishMessage
                        .variableHeader()
                        .packetId());
                /*推送消息*/
                MqttPublishMessage usrPlcentPublishMessage = new MqttPublishMessage(Header, publishVariableHeader, sendBuf);
                /*推送到MQ处理*/
                sendToMQ(usrPlcentPublishMessage, clientId);
            }
            //自定义订阅主题
            else if (topicName.endsWith(TopicType.FUNCTION_GET.getTopicSuffix()) || topicName.endsWith(TopicType.PROPERTY_GET.getTopicSuffix())) {
                log.debug("开始处理自定义订阅主题");
                //然后转发消息到订阅自定义主题的客户端,自定义主题为空不执行
                String topic = deviceMqttConnect.getDyname();
                if (StringUtils.isNotBlank(topic)) {
                    MqttQoS qos = publishMessage.fixedHeader().qosLevel();
                    ByteBuf sendBuf = publishMessage.content().retainedDuplicate();
                    sendBuf.resetReaderIndex();
                    /*配置推送消息类型*/
                    MqttFixedHeader Header = new MqttFixedHeader(MqttMessageType.PUBLISH, false, qos, publishMessage
                            .fixedHeader()
                            .isRetain(), 0);
                    /*设置topic packetId*/
                    MqttPublishVariableHeader publishVariableHeader = new MqttPublishVariableHeader(topic, publishMessage
                            .variableHeader()
                            .packetId());
                    /*推送消息*/
                    MqttPublishMessage usrPlcentPublishMessage = new MqttPublishMessage(Header, publishVariableHeader, sendBuf);
                    /*推送到订阅的客户端*/
                    sendMessageToClients(usrPlcentPublishMessage);
                    log.debug("=>***客户端[{}],主题[{}],推送消息[{}]", clientId, topic, ByteBufUtil.hexDump(publishMessage.content()));
                } else {
                    log.warn("自定义订阅主题为空，未转发到自定义主题!,消息内容：[{}]", ByteBufUtil.hexDump(publishMessage.content()));
                }
            }
        }
    }

    /**
     * 原发布和订阅逻辑
     */
    public void MqttYllj(ChannelHandlerContext ctx, String clientId, MqttPublishMessage publishMessage) {
        /*获取客户端session*/
        Session session = AttributeUtils.getSession(ctx.channel());
        //平台检测session是否同步
        boolean containsKey = sessionStore.containsKey(clientId);
        if (!containsKey) {
            SessionManger.buildSession(clientId, session);
        }
        /*推送保留信息*/
        pubRetain(publishMessage);
        /*响应客户端消息到达Broker*/
        callBack(session, publishMessage, clientId);
        /*推送到订阅的客户端*/
        sendMessageToClients(publishMessage);
        /*推送到MQ处理*/
        sendToMQ(publishMessage, clientId);
        /*累计接收消息数*/
        redisCache.incr2(YanfanConstant.REDIS.MESSAGE_RECEIVE_TOTAL, -1L);
        redisCache.incr2(YanfanConstant.REDIS.MESSAGE_RECEIVE_TODAY, 60 * 60 * 24);
    }

    /**
     * 消息推送
     *
     * @param message 推送消息
     */
    @SneakyThrows
    public void sendToMQ(MqttPublishMessage message, String clientId) {
        /*获取topic*/
        String topicName = message.variableHeader().topicName();
        byte[] source = ByteBufUtil.getBytes(message.content());
        DeviceReportBo reportBo = DeviceReportBo
                .builder()
                .serialNumber(topicsUtils.parseSerialNumber(topicName))
                .topicName(topicName)
                .packetId((long) message.variableHeader().packetId())
                .platformDate(DateUtils.getNowDate())
                .data(source)
                .serverType(ServerType.MQTT)
                .build();
        if (topicName.endsWith(YanfanConstant.TOPIC.MSG_REPLY) || topicName.endsWith(YanfanConstant.TOPIC.UPGRADE_REPLY)) {
            /*设备应答服务器回调数据*/
            reportBo.setReportType(2);
        } else {
            /*设备上报数据*/
            reportBo.setReportType(1);
        }
        // 规则引擎脚本处理,完成后返回结果
        MsgContext context = ruleProcess.processRuleScript(topicsUtils.parseSerialNumber(topicName), 1, topicName, new String(source));
        if (!Objects.isNull(context) && StringUtils.isNotEmpty(context.getPayload()) && StringUtils.isNotEmpty(context.getTopic())) {
            reportBo.setTopicName(context.getTopic());
            reportBo.setData(context.getPayload().getBytes(StandardCharsets.UTF_8));
        }

        if (reportBo.getTopicName().contains("property")) {
            MessageProducer.sendPublishMsg(reportBo);
        } else if (reportBo.getTopicName().contains("status")) {
            String jsonString = new String(reportBo.getData(), StandardCharsets.UTF_8);
            JSONObject jsonObject = JSONObject.parseObject(jsonString);
            int status = jsonObject.getInteger("status");
            DeviceStatusBo bo = DeviceStatusBo
                    .builder()
                    .serialNumber(topicsUtils.parseSerialNumber(reportBo.getTopicName()))
                    .status(DeviceStatus.convert(status))
                    .build();
            MessageProducer.sendStatusMsg(bo);
        } else {
            MessageProducer.sendOtherMsg(reportBo);
        }

    }

    /**
     * 发送模拟数据进行处理
     *
     * @param message
     */
    public void sendTestToMQ(MqttPublishMessage message) {
        /*获取topic*/
        String topicName = message.variableHeader().topicName();
        DeviceReportBo reportBo = DeviceReportBo
                .builder()
                .serialNumber(topicsUtils.parseSerialNumber(topicName))
                .topicName(topicName)
                .packetId((long) message.variableHeader().packetId())
                .platformDate(DateUtils.getNowDate())
                .data(ByteBufUtil.getBytes(message.content()))
                .build();
        MessageProducer.sendOtherMsg(reportBo);
    }


    /**
     * 推送消息到订阅客户端
     *
     * @param message 消息
     */
    public void sendMessageToClients(MqttPublishMessage message) {
        ClientManager.pubTopic(message);
    }


    /**
     * 应答客户端，消息到达Broker
     *
     * @param session 客户端
     * @param message 消息
     */
    private void callBack(Session session, MqttPublishMessage message, String clientId) {
        /*获取消息等级*/
        MqttQoS mqttQoS = message.fixedHeader().qosLevel();
        int packetId = message.variableHeader().packetId();
        MqttFixedHeader header;
        switch (mqttQoS.value()) {
            /*0,1消息等级，直接回复*/
            case 0:
            case 1:
                header = new MqttFixedHeader(MqttMessageType.PUBACK, false, mqttQoS, false, 0);
                break;
            case 2:
                // 处理Qos2的消息确认
                if (!messageStore.outRelContains(packetId)) {
                    messageStore.saveRelInMsg(packetId);
                }
                header = new MqttFixedHeader(MqttMessageType.PUBREC, false, MqttQoS.AT_MOST_ONCE, false, 0);
                break;
            default:
                header = null;
        }
        /*处理消息等级*/
        handleMqttQos(packetId, mqttQoS, true, clientId);
        /*响应客户端*/
        MqttMessageIdVariableHeader variableHeader = null;
        if (packetId > 0) {
            variableHeader = MqttMessageIdVariableHeader.from(packetId);
        }
        MqttPubAckMessage ackMessage = new MqttPubAckMessage(header, variableHeader);
        if (mqttQoS.value() >= 1) {
            ResponseManager.responseMessage(session, ackMessage, true);
        }
        /*更新客户端ping时间*/
        ClientManager.updatePing(session.getClientId());

    }

    /**
     * Qos不同消息处理
     */
    private void handleMqttQos(int packetId, MqttQoS qoS, boolean clearSession, String clientId) {
        if (qoS == MqttQoS.AT_LEAST_ONCE || qoS == MqttQoS.EXACTLY_ONCE) {
            ClientMessage clientMessage = ClientMessage.of(clientId, qoS, null, false);
            messageStore.savePubMsg(packetId, clientMessage);
        }
    }


    /**
     * 推送保留信息
     */
    @SneakyThrows
    private void pubRetain(MqttPublishMessage message) {
        redisCache.incr2(YanfanConstant.REDIS.MESSAGE_RETAIN_TOTAL, -1L);
        /*根据message.fixedHeader().isRetain() 判断是否有保留信息*/
        RetainMsgManager.pushMessage(message);
    }


}
